package cn.data.stream.stream;

import cn.data.stream.enums.StreamType;
import cn.data.stream.expection.DataStreamInitException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;

/**
 * datastream管理器 可获取所有stream
 *
 * @author shi hong yu
 */
@Slf4j
public class DataStreamManager {

    private List<IComponent> inputComponentList = new ArrayList<>();
    private List<IComponent> streamComponentList = new ArrayList<>();
    private List<IComponent> outputComponentList = new ArrayList<>();

    private Map<String, IComponent> componentMap = new HashMap<>();

    public void addComponent(StreamType type, IComponent component) {
        if (component == null) {
            return;
        }
        if (componentMap.containsKey(component.getName())) {
            return;
        }
        componentMap.put(component.getName(), component);
        log.info("添加data-stream到manager name:{} class:{}",component.getName(),component.getClass());
        switch (type) {
            case INPUT:
                inputComponentList.add(component);
                break;
            case STREAM:
                streamComponentList.add(component);
                break;
            case OUTPUT:
                outputComponentList.add(component);
                break;
        }
    }

    public IComponent getComponent(String componentName) {
        return componentMap.get(componentName);
    }

    public boolean containsComponent(String componentName) {
        return componentMap.containsKey(componentName);
    }

    public void init() throws DataStreamInitException {
        log.info("开始初始化output-stream");
        for (IComponent comp : outputComponentList) {
            comp.init();
        }
        log.info("output-stream 完成初始化数目 {}", outputComponentList.size());

        log.info("开始初始化stream");
        for (IComponent comp : streamComponentList) {
            comp.init();
        }
        log.info("stream 完成初始化数目 {}", streamComponentList.size());

        log.info("开始初始化input-stream");
        for (IComponent comp : inputComponentList) {
            comp.init();
        }
        log.info("input-stream 完成初始化数目 {}", inputComponentList.size());
    }

    @PreDestroy
    public void destroy() {
        log.info("开始销毁input-stream");
        for (IComponent comp : inputComponentList) {
            comp.destroy();
        }
        log.info("input-stream 完成销毁数目 {}", inputComponentList.size());

        log.info("开始销毁stream");
        for (IComponent comp : streamComponentList) {
            comp.destroy();
        }
        log.info("stream 完成销毁数目 {}", streamComponentList.size());

        log.info("开始销毁output-stream");
        for (IComponent comp : outputComponentList) {
            comp.destroy();
        }
        log.info("output-stream 完成销毁数目 {}", outputComponentList.size());
    }

    public void prettyPrint() {
        //TODO 以图的形式展示stream配置关系
    }
}
